package com.wyl.es.config;

import lombok.Data;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.bulk.*;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;
import java.util.function.BiConsumer;

import static org.elasticsearch.action.bulk.BulkProcessor.builder;

/**
 * @author: wyl
 * @date: 2021/8/5
 * @des:
 */
@Data
@Configuration
@ConfigurationProperties(prefix = "elasticsearch")
public class EsConfig {

    private String host;
    private Integer port;
    private String user;
    private String pwd;

    @Bean(destroyMethod = "close")
    public RestHighLevelClient client() {

        HttpHost httpHost = new HttpHost(host, port, "http");

        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, pwd));

        RestClientBuilder builder = RestClient.builder(httpHost).setRequestConfigCallback(requestConfigBuilder -> {
            requestConfigBuilder.setConnectTimeout(-1);
            requestConfigBuilder.setSocketTimeout(-1);
            requestConfigBuilder.setConnectionRequestTimeout(-1);
            return requestConfigBuilder;
        }).setHttpClientConfigCallback(httpClientBuilder -> {
            httpClientBuilder.disableAuthCaching();
            return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
        });

        RestHighLevelClient client = new RestHighLevelClient(builder);
        return client;

    }

    @Bean
    public BulkProcessor bulkProcessor() {
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            //beforeBulk会在批量提交之前执行
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
//                System.out.println("调用procss bulk 之前" + executionId +" items info:" + executionId);
            }

            //第一个afterBulk会在批量成功后执行，可以跟beforeBulk配合计算批量所需时间
            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                failure.printStackTrace();
                List<DocWriteRequest<?>> list = request.requests();
            }

            //第二个afterBulk会在批量失败后执行
            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
//                System.out.println("调用procss bulk 之后" + executionId +" items info:" + response.getItems());
            }
        };

        BulkProcessor bulkProcessor = BulkProcessor.builder(
                (request, bulkListener) -> (client()).bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
                listener)
                // 每添加500个request，执行一次bulk操作
                .setBulkActions(500)
                // 每达到1M 执行一次
                .setBulkSize(new ByteSizeValue(1, ByteSizeUnit.MB))
                // 每1s执行一次bulk操作
                .setFlushInterval(TimeValue.timeValueSeconds(1))
                // 0代表同步提交即只能提交一个request 10代表当有一个新的bulk正在累积时，10个并发请求可被允许执行
                .setConcurrentRequests(10)
                // 设置当出现代表ES集群拥有很少的可用资源来处理request时抛出
                // EsRejectedExecutionException造成N个bulk内request失败时
                // 进行重试的策略,初始等待100ms，后面指数级增加，总共重试3次.
                // 不重试设为BackoffPolicy.noBackoff()
                .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(1), 3)).build();
        return bulkProcessor;
    }

}